package com.zlzlib.librxoknet.util;


import android.os.SystemClock;

import com.zlzlib.librx.BaseProgressObserver;
import com.zlzlib.librx.ProgressInfo;
import com.zlzlib.librx.listener.ProgressListener;
import com.zlzlib.librxoknet.function.FileFunction;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.observers.DisposableObserver;
import io.reactivex.rxjava3.schedulers.Schedulers;
import okhttp3.ResponseBody;

/**
 * Created by zlz on 2019/8/13.
 */
public class RxUtil {

    //不用转化直接得到
    public static <T> Observer<T> toSubscribe(Observable<T> o, Observer<T> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }

    public static <T> DisposableObserver<T> toSubscribe(Observable<T> o, DisposableObserver<T> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }

    //T 转化为R
    public static <T, R> Observer<R> toSubscribe(Observable<T> o, Function<T, R> f, Observer<R> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                //转换器在子线程转换
                .map(f)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }


    public static <T, R> DisposableObserver<R> toSubscribe(Observable<T> o, Function<T, R> f, Consumer<? super R> onNext, DisposableObserver<R> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                //转换器在子线程转换
                .map(f)
                .observeOn(Schedulers.io())
                .doOnNext(onNext)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }


    public static <T, R> DisposableObserver<R> toSubscribe(Observable<T> o, Function<T, R> f, DisposableObserver<R> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                //转换器在子线程转换
                .map(f)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }

    //T 转化为R  中间可以增加 ObservableSource 的转化
    public static <T, R> Observer<R> toSubscribe(Observable<T> o, Consumer<T> c, Function<T, ObservableSource<R>> f, Observer<R> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .doOnNext(c)
                .observeOn(Schedulers.io())
                //转换器在子线程转换
                .flatMap(f)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }

    public static <T, R> DisposableObserver<R> toSubscribe(Observable<T> o, Consumer<T> c, Function<T, ObservableSource<R>> f, DisposableObserver<R> s) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .doOnNext(c)
                .observeOn(Schedulers.io())
                //转换器在子线程转换
                .flatMap(f)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(s);
    }

    /**
     * 下载文件
     *
     * @param o        下载体
     * @param path     本地存储地址
     * @param fileName 文件名称
     * @param s        回调  进度监听
     * @return
     */
    public static BaseProgressObserver<File> downFileByProgress(Observable<ResponseBody> o, String path, String fileName, BaseProgressObserver<File> s) {
        return downFileByProgress(o, s, path, fileName).subscribeWith(s);
    }

    /**
     * 下载文件
     *
     * @param o        下载体
     * @param listener 进度监听
     * @param path     本地存储地址
     * @param fileName 文件名称
     * @param s        回调
     * @return
     */
    public static Observer<File> downFileByProgress(Observable<ResponseBody> o, String path, String fileName, ProgressListener listener, Observer<File> s) {
        return downFileByProgress(o, listener, path, fileName).subscribeWith(s);
    }

    /**
     * 下载文件
     *
     * @param o        下载体
     * @param listener 进度监听
     * @param path     本地存储地址
     * @param fileName 文件名称
     * @param s        回调
     * @return
     */
    public static Disposable downFileByProgress(Observable<ResponseBody> o, String path, String fileName, ProgressListener listener, Consumer<File> s) {
        return downFileByProgress(o, listener, path, fileName).subscribe(s);
    }

    /**
     * 下载文件
     *
     * @param o        下载体
     * @param listener 进度监听
     * @param path     本地存储地址
     * @param fileName 文件名称
     * @return
     */
    public static Observable<File> downFileByProgress(Observable<ResponseBody> o, ProgressListener listener, String path, String fileName) {
        return o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(Schedulers.computation())
                .map(new FileFunction(listener, path, fileName))
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 存储文件到磁盘
     *
     * @param is           文件流
     * @param info         进度数据
     * @param listener     监听回调
     * @param destFileDir  文件存储地址
     * @param destFileName 文件名称
     * @param isAppendFile 是否拼接写入
     * @return
     * @throws Exception
     */
    public static File saveFile(InputStream is, final ProgressInfo info, final ProgressListener listener,
                                String destFileDir, String destFileName, boolean isAppendFile) throws Exception {
        byte[] buf = new byte[4096];
        int len = 0;
        FileOutputStream fos = null;
        try {
            //当前数据总大小
            long sum = 0;
            File dir = new File(destFileDir);
            if (!dir.exists()) {
                dir.mkdirs();
            }
            File file = new File(dir, destFileName);
            //追加还是覆盖  是否支持断点下载
            fos = new FileOutputStream(file, isAppendFile);
            long lastRefreshTime = SystemClock.elapsedRealtime();  //最后一次刷新的时间
            long tempSize = 0L;
            while ((len = is.read(buf)) != -1) {
                sum += len;
                tempSize += len;
                fos.write(buf, 0, len);
                if (null != listener) {
                    long curTime = SystemClock.elapsedRealtime();
                    long intervalTime = curTime - lastRefreshTime;
                    if (intervalTime >= 500) {
                        info.setEachBytes(sum == -1 ? -1 : tempSize);
                        info.setCurrentBytes(sum);
                        info.setIntervalTime(intervalTime);
                        //发送到主线程
                        AndroidSchedulers.mainThread().scheduleDirect(new Runnable() {
                            @Override
                            public void run() {
                                listener.updateDown(info);
                            }
                        });
                        tempSize = 0;
                        lastRefreshTime = curTime;
                    }
                }
            }
            info.setCurrentBytes(sum);
            info.setFinish(true);
            if (null != listener) {
                AndroidSchedulers.mainThread().scheduleDirect(new Runnable() {
                    @Override
                    public void run() {
                        listener.updateDown(info);
                    }
                });
            }
            fos.flush();
            return file;
        } finally {
            try {
                if (is != null) is.close();
            } catch (IOException e) {
            }
            try {
                if (fos != null) fos.close();
            } catch (IOException e) {
            }
        }
    }
}
